package com.atguigu.sparkstreaming.examples

import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import org.apache.spark.streaming.kafka010.{CanCommitOffsets, HasOffsetRanges, KafkaUtils, OffsetRange}
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * Created by Smexy on 2022/7/15
 *
 * Exception in thread "main" java.lang.ClassCastException:
 *    org.apache.spark.streaming.dstream.MappedDStream
 *        cannot be cast to
 *    org.apache.spark.streaming.kafka010.CanCommitOffsets
 *
 *
 *    只有初始DS能提交偏移量！ 只有初始DS是DirectKafkaInputDStream，只有
 *        DirectKafkaInputDStream才能 asInstanceOf[CanCommitOffsets]
 *
 *
 */
object CommitOffsetsDemo {

  def main(args: Array[String]): Unit = {

    val streamingContext = new StreamingContext("local[*]", "TransformDemo", Seconds(10))

    //所有的消费者参数都可以在 ConsumerConfig
    val kafkaParams = Map[String, Object](
      "bootstrap.servers" -> "hadoop102:9092,hadoop103:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> "2203092",
      "auto.offset.reset" -> "latest",
      "enable.auto.commit" -> "false"
    )


    val topics = Array("topicA")

    // ds: DirectKafkaInputDStream  会每10s采到的数据封装为 KafkaRDD
    val ds: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String](
      streamingContext,
      PreferConsistent,
      Subscribe[String, String](topics, kafkaParams)
    )

    //在Driver端
    var ranges: Array[OffsetRange] = null
    //获取偏移量
    val ds1: DStream[ConsumerRecord[String, String]] = ds.transform(rdd => {

      //偏移量  MapPartitionsRDD
      ranges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges

      rdd

    })

    // ds2:MappedDStream
    val ds2: DStream[String] = ds1.map(record => record.value())


    ds2.foreachRDD(rdd => {

      //输出  在Executor端
      rdd.foreach(word => println(Thread.currentThread().getName + ":"+word))

      //手动提交offsets
      ds2.asInstanceOf[CanCommitOffsets].commitAsync(ranges)

    })

    // 启动APP
    streamingContext.start()

    // 阻塞进程，让进程一直运行
    streamingContext.awaitTermination()

  }

}
